import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.ForeachWriter

object MysqlWriter {
  val url = "jdbc:mysql://linux123:3306/db_businfo?characterEncoding=UTF-8&useSSL=false"
  val user = "hive"
  val password = "12345678"
  Class.forName("com.mysql.jdbc.Driver")
}

class MysqlWriter extends ForeachWriter[BusInfo] {
  var con: Connection = _
  var ps: PreparedStatement = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    con = DriverManager.getConnection(MysqlWriter.url, MysqlWriter.user, MysqlWriter.password)
    true
  }


  override def process(info: BusInfo): Unit = {
    val lglatStrs = info.lglat.split("_")
    val longitude = lglatStrs(0)
    val latitude = lglatStrs(1)
    ps = con.prepareStatement("insert into bus_info values (?,?)")
    ps.setString(1, longitude)
    ps.setString(2, latitude)

    ps.executeUpdate()

  }

  override def close(errorOrNull: Throwable): Unit = {
    if (con != null) {
      con.close()
    }
    if (ps != null) {
      ps.close()
    }
    if (errorOrNull != null) errorOrNull.printStackTrace()

  }
}
